当我们讨论 Netty线程模型的时候,一般首先会想到的是经典的 Reactor线程模型,尽管不同的NIO框架对于 Reactor模式的实现存在差异,但本质上还是遵循了 Reactor的基础线程模型。 下面让我们一起回顾经典的 Reactor线程模型。

Reactor单线程模型,是指所有的IO操作都在同一个NIO线程上面完成。NIO线程的职责如下。

Reactor单线程模型如图18-1所示。

由于 Reactor模式使用的是异步非阻塞IO,所有的IO操作都不会导致阻塞,理论上一个线程可以独立处理所有IO相关的操作。从架构层面看,一个NIO线程确实可以完成其承担的职责。例如,通过 Acceptor类接收客户端的TCP连接请求消息,当链路建立成功之后,通过 Dispatch将对应的 ByteBuffer派发到指定的 Handler上,进行消息解码。用户线程消息编码后通过NIO线程将消息发送给客户端。 在一些小容量应用场景下,可以使用单线程模型。但是这对于高负载、大并发的应用场景却不合适,主要原因如下。

为了解决这些问题,演进出了 Reactor多线程模型。下面我们一起学习下 Reactor多线程模型。

Rector多线程模型与单线程模型最大的区别就是有一组NIO线程来处理IO操作,它的原理如图18-2所示。 Reactor多线程模型的特点如下

在绝大多数场景下, Reactor多线程模型可以满足性能需求。但是,在个别特殊场景中,一个NIO线程负责监听和处理所有的客户端连接可能会存在性能问题。例如并发百万客户端连接,或者服务端需要对客户端握手进行安全认证,但是认证本身非常损耗性能在这类场景下,单独一个 Acceptor线程可能会存在性能不足的问题,为了解决性能问题,产生了第三种 Reactor线程模型——主从 Reactor多线程模型。

主从 Reactor线程模型的特点是:服务端用于接收客户端连接的不再是一个单独的NIO线程,而是一个独立的NIO线程池。 Acceptor接收到客户端TCP连接请求并处理完成后(可能包含接入认证等),将新创建的 SocketChannel注册到IO线程池(sub reactor线程池)的某个IO线程上,由它负责 SocketChannel的读写和编解码工作。 Acceptor线程池仅仅用于客户端的登录、握手和安全认证,一旦链路建立成功,就将链路注册到后端 sub reactor线程池的IO线程上,由IO线程负责后续的O操作。 它的线程模型如图18-3所示。

利用主从NIO线程模型,可以解决一个服务端监听线程无法有效处理所有客户端连接的性能不足问题。因此,在Netty的官方Demo中,推荐使用该线程模型。

Netty的线程模型并不是一成不变的,它实际取决于用户的启动参数配置。通过设置不同的启动参数,Netty可以同时支持 Reactor单线程模型、多线程模型和主从 Reactor多线层模型。 下面让我们通过一张原理图(图18-4)来快速了解Netty的线程模型。

可以通过如图18-5所示的Netty服务端启动代码来了解它的线程模型。

服务端启动的时候,创建了两个 NioEventLoopGroup,它们实际是两个独立的 Reactor线程池。一个用于接收客户端的TCP连接,另一个用于处理IO相关的读写操作,或者执行系统Task、定时任务Task等。 Netty用于接收客户端请求的线程池职责如下。 (1)接收客户端TCP连接,初始化 Channel参数; (2)将链路状态变更事件通知给 ChannelPipeline。 Netty处理IO操作的 Reactor线程池职责如下。 (1)异步读取通信对端的数据报,发送读事件到 ChannelPipeline; (2)异步发送消息到通信对端,调用 ChannelPipeline的消息发送接口; (3)执行系统调用Task; (4)执行定时任务Task,例如链路空闲状态监测定时任务。 通过调整线程池的线程个数、是否共享线程池等方式,Netty的 Reactor线程模型可以在单线程、多线程和主从多线程间切换,这种灵活的配置方式可以最大程度地满足不同用户的个性化定制。 为了尽可能地提升性能, Netty在很多地方进行了无锁化的设计,例如在IO线程内部进行串行操作,避免多线程竞争导致的性能下降问题。表面上看,串行化设计似乎CPU利用率不高,并发程度不够。但是,通过调整NIO线程池的线程参数,可以同时启动多个串行化的线程并行运行,这种局部无锁化的串行线程设计相比一个队列一多个工作线程的模型性能更优。 它的设计原理如图18-6所示

Netty的 NioEventLoop读取到消息之后,直接调用 ChannelPipeline的fireChannelRead(Object msg)。只要用户不主动切换线程,一直都是由 NioEventLoop调用用户的 Handler,期间不进行线程切换。这种串行化处理方式避免了多线程操作导致的锁的竞争,从性能角度看是最优的。

Netty的多线程编程最佳实践如下 (1)创建两个 NioEventLoopGroup,用于逻辑隔离 NIO Acceptor和 NIO I/O线程。 (2)尽量不要在 ChannelHandler中启动用户线程(解码后用于将POJO消息派发到后端业务线程的除外)。 (3)解码要放在NIO线程调用的解码 Handler中进行,不要切换到用户线程中完成消息的解码。 (4)如果业务逻辑操作非常简单,没有复杂的业务逻辑计算,没有可能会导致线程被阻塞的磁盘操作、数据库操作、网路操作等,可以直接在NIO线程上完成业务逻辑编排,不需要切换到用户线程。 5)如果业务逻辑处理复杂,不要在NIO线程上完成,建议将解码后的POJO消息封装成Task,派发到业务线程池中由业务线程执行,以保证NIO线程尽快被释放,处理其他的IO操作。 推荐的线程数量计算公式有以下两种 ◎公式一:线程数量=(线程总时间/瓶颈资源时间)×瓶颈资源的线程并行数 ◎公式二:QPS=1000/线程总时间×线程数 由于用户场景的不同,对于一些复杂的系统,实际上很难计算出最优线程配置,只能是根据测试数据和用户场景,结合公式给出一个相对合理的范围,然后对范围内的数据进行性能测试,选择相对最优值。

Netty的 NioEventLoop并不是一个纯粹的IO线程,它除了负责IO的读写之外,还兼顾处理以下两类任务。 ◎系统Task:通过调用 NioEventLoop的 execute(Runnable task)方法实现,Netty有很多系统Task,创建它们的主要原因是:当IO线程和用户线程同时操作网络资源时,为了防止并发操作导致的锁竞争,将用户线程的操作封装成Task放入消息队列中,由IO线程负责执行,这样就实现了局部无锁化。 ◎定时任务:通过调用 NioEventLoop的 schedule(Runnable command, long delay,TimeUnit unit)方法实现。 正是因为 NioEventLoop具备多种职责,所以它的实现比较特殊,它并不是个简单的Runnable。

它实现了 EventLoop接口、 EventExecutorGroup接口和ScheduledService接口,正是因为这种设计,导致 NioEventLoop和其父类功能实现非常复杂。下面我们就重点分析它的源码实现,理解它的设计原理。

从下个小节开始,我们将对NioEventLoop的源码进行分析。通过源码分析,希望读者能够理解Netty的Reactor线程设计原理,掌握其精髓。 

作为NIO框架的 Reactor线程, NioEventLoop需要处理网络IO读写事件,因此它必须聚合一个多路复用器对象。下面我们看它的Selector定义,如图18-9所示。 Selector的初始化非常简单,直接调用 Selector.open()方法就能创建并打开一个新的Selector。Netty对 Selector的 selectedKeys进行了优化,用户可以通过io.nett.noKeySetOptimization开关决定是否启用该优化项。默认不打开 selectedKeys优化功能。 Selector的初始化代码如图18-10所示。

如果没有开启 selectedKeys优化开关,通过 provider.openSelector()创建并打开多路复用器之后就立即返回。 如果开启了优化开关,需要通过反射的方式从 Selector实例中获取 selectedKeys和publicSelectedKeys,将上述两个成员变量设置为可写,通过反射的方式使用 Netty构造的selectedKeys包装类 selectedKeySet将原JDK的 selectedKeys替换掉。 分析完 Selector的初始化,下面重点看下run方法的实现,如图18-11所示。 所有的逻辑操作都在for循环体内进行,只有当 NioEventLoop接收到退出指令的时候,才退出循环,否则一直执行下去,这也是通用的NIO线程实现方式。 首先需要将 wakenUp还原为 false,并将之前的 wakeUp状态保存到 oldWakenUp变量中。通过 hasTasks()方法判断当前的消息队列中是否有消息尚未处理,如果有则调用selectNow()方法立即进行一次 select操作,看是否有准备就绪的 Channel需要处理。它的代码实现如图18-12所示。

Selector的 selectNow()方法会立即触发 Selector的选择操作,如果有准备就绪的Channel,则返回就绪 Channel的集合,否则返回0。选择完成之后,再次判断用户是否调用了 Selector的 wakeup()方法,如果调用,则执行 selector.wakeup()操作。 下面我们返回到run方法,继续分析代码。如果消息队列中没有消息需要处理,则执行 select()方法,由 Selector多路复用器轮询,看是否有准备就绪的 Channel。它的实现如图18-13所示。 取当前系统的纳秒时间,调用 delayNanos()方法计算获得 NioEventLoop中定时任务的触发时间。

计算下一个将要触发的定时任务的剩余超时时间,将它转换成毫秒,为超时时间增加0.5毫秒的调整值。对剩余的超时时间进行判断,如果需要立即执行或者已经超时,则调用 selector.selectNow()进行轮询操作,将 selectCnt设置为1,并退出当前循环将定时任务剩余的超时时间作为参数进行 select操作,每完成一次 select操作,对 select计数器 selectCnt加1。 Select操作完成之后,需要对结果进行判断,如果存在下列任意一种情况,则退出当前循环。

如果本次 Selector的轮询结果为空,也没有 wakeup操作或是新的消息需要处理,则说明是个空轮询,有可能触发了JDK的 epoll bug,它会导致 Selector的空轮询,使I/O线程一直处于100%状态。截止到当前最新的JDK7版本,该bug仍然没有被完全修复。所以Netty需要对该bug进行规避和修正。 Bug-id=6403933的 Selector堆栈如图18-14所示。 该Bug的修复策略如下。 (1)对 Selector的 select操作周期进行统计; (2)每完成一次空的 select操作进行一次计数 (3)在某个周期(例如100ms)内如果连续发生N次空轮询,说明触发了 JDK NIO的 epoll()死循环bug。

监测到 Selector处于死循环后,需要通过重建 Selector的方式让系统恢复正常,代码如图18-15所示。

首先通过 inEventLoopo方法判断是否是其他线程发起的 rebuildSelector,如果由其他线程发起,为了避免多线程并发操作 Selector和其他资源,需要将 rebuildSelector封装成Task,放到 NioEventLoop的消息队列中,由 NioEventLoop线程负责调用,这样就避免了多线程并发操作导致的线程安全问题调用 openSelector方法创建并打开新的 Selector,通过循环,将原 Selector上注册的SocketChannel从旧的 Selector上去注册,重新注册到新的 Selector上,并将老的 Selector关闭。 相关代码如图18-16所示。

通过销毁旧的、有问题的多路复用器,使用新建的 Selector,就可以解决空轮询 Selector导致的I/O线程CPU占用100%的问题。 如果轮询到了处于就绪状态的 SocketChannel,则需要处理网络IO事件,相关代码如图18-17所示。

由于默认未开启 selectedKeys优化功能,所以会进入 processSelectedKeysPlain分支执行。下面继续分析 processSelectedKeysPlain的代码实现,如图18-18所示。

对 SelectionKey进行保护性判断,如果为空则返回。获取 SelectionKey的迭代器进行循环操作,通过迭代器获取 SelectionKey和 SocketChannel的附件对象,将已选择的选择键从迭代器中删除,防止下次被重复选择和处理,代码如图18-19所示。

对 SocketChannel的附件类型进行判读,如果是AbstractNioChannel类型,说明它是NioServerSocketChannel或者NioSocketChannel,需要进行IO读写相关的操作;如果它是NioTask,则对其进行类型转换,调用 processSelectedKey进行处理。由于Netty自身没实现 NioTask接口,所以通常情况下系统不会执行该分支,除非用户自行注册该Task到多路复用器。 下面我们继续分析IO事件的处理,代码如图18-20所示。

首先从 NioServerSocketChanne或者 NioSocketChannel中获取其内部类 Unsafe,判断当前选择键是否可用,如果不可用,则调用 Unsafe的 close方法,释放连接资源。如果选择键可用,则继续对网络操作位进行判断,代码如图18-21所示。

如果是读或者连接操作,则调用Unsafe的read方法。此处Unsafe的实现是个多态,对于 NioServerSocketchannel,它的读操作就是接收客户端的TCP连接,相关代码如图18-22所示。

对于 NioSocketChannel,它的读操作就是从 SocketChannel中读取 ByteBuffer,相关代码如图18-23所示。 如果网络操作位为写,则说明有半包消息尚未发送完成,需要继续调用fush方法进行发送,相关的代码如图18-24所示。 如果网络操作位为连接状态,则需要对连接结果进行判读,代码如图18-25所示。

需要注意的是,在进行 finishConnect判断之前,需要将网络操作位进行修改,注销掉SelectionKey.OP_CONNECT。 处理完IO事件之后, NioEventLoop需要执行非IO操作的系统Task和定时任务,代码如图18-26所示。 由于 NioEventLoop需要同时处理IO事件和非IO任务,为了保证两者都能得到足够的CPU时间被执行,Netty提供了lO比例供用户定制。如果lO操作多于定时任务和Task,则可以将IO比例调大,反之则调小,默认值为50%。 Task的执行时间根据本次IO操作的执行时间计算得来。下面我们具体看 runAllTasks方法的实现,如图18-27所示。 首先从定时任务消息队列中弹出消息进行处理,如果消息队列为空,则退出循环。根据当前的时间戳进行判断,如果该定时任务已经或者正处于超时状态,则将其加入到执行TaskQueue中,同时从延时队列中删除。定时任务如果没有超时,说明本轮循环不需要处理,直接退出即可,代码实现如图18-28所示。

执行 TaskQueue中原有的任务和从延时队列中复制的已经超时或者正处于超时状态的定时任务,代码如图18-29所示。 由于获取系统纳秒时间是个耗时的操作,每次循环都获取当前系统纳秒时间进行超时判断会降低性能。为了提升性能,每执行60次循环判断一次,如果当前系统时间已经到了分配给非IO操作的超时时间,则退出循环。这是为了防止由于非IO任务过多导致IO操作被长时间阻塞。

最后,判断系统是否进入优雅停机状态,如果处于关闭状态,则需要调用 closeAll方法,释放资源,并让 NioEventLoop线程退岀循环,结束运行。资源关闭的代码实现如图18-30所示。

遍历获取所有的 Channel,调用它的 Unsafe.closed方法关闭所有链路,释放线程池、ChannelPipeline和 ChannelHandler等资源。